package edu.algorithm.knn.parallel.impl;

import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;
import java.util.concurrent.RecursiveAction;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.algorithm.knn.parallel.common.DataPoint;
import edu.algorithm.knn.parallel.common.Utils;

/**
 * 实现RecursiveAction
 * @author liukun
 *
 */
public class KNN extends RecursiveAction {

	private static final long serialVersionUID = -991721883833384927L;

	private static final Logger LOGGER = LoggerFactory.getLogger(KNN.class);

	private DataPoint[] testDatas;
	private int begin;
	private int end;
	private List<DataPoint> trainDataList;
	private int K;
	private int threshold = 1000;

	public KNN(DataPoint[] testDatas, int start, int end, List<DataPoint> trainDataList, int K) {
		this.testDatas = testDatas;
		this.begin = start;
		this.end = end;
		this.trainDataList = trainDataList;
		this.K = K;
	}
	
	public KNN(DataPoint[] testDatas, int start, int end, List<DataPoint> trainDataList, int K,int threshold) {
		this.testDatas = testDatas;
		this.begin = start;
		this.end = end;
		this.trainDataList = trainDataList;
		this.K = K;
		this.threshold = threshold; 
	}

	@Override
	protected void compute() {
		// 如果任务足够小，就执行计算
		if (end - begin <= threshold) {
			LOGGER.info("开始计算start is {}, end is {}", begin, end);
			long begintime = System.currentTimeMillis();
			for (int i = begin; i <= end; i++) {
				DataPoint testPoint = testDatas[i];
				PriorityQueue<DataPoint> priorityQueue = new PriorityQueue<>(K, new Comparator<DataPoint>() {
					@Override
					public int compare(DataPoint o1, DataPoint o2) {
						float dis1 = Utils.similarScore(o1, testPoint);
						float dis2 = Utils.similarScore(o2, testPoint);
						if (dis1 < dis2) {
							return -1;
						} else {
							return 1;
						}
					}
				});
				for (DataPoint trainDataPoint : trainDataList) {
					if (priorityQueue.size() < K) {
						priorityQueue.add(trainDataPoint);
					} else {
						DataPoint top = priorityQueue.peek();
						if (Utils.similarScore(top, testPoint) > Utils.similarScore(trainDataPoint, testPoint)) {
							priorityQueue.poll();
							priorityQueue.add(trainDataPoint);
						}
					}
				}
			}
			long endtime = System.currentTimeMillis();
			LOGGER.info("计算结束用时：{}", endtime - begintime);
			return;
		} else {
			int middle = (end + begin) / 2;
			KNN task1 = new KNN(testDatas, begin, middle, trainDataList, K);
			KNN task2 = new KNN(testDatas, middle + 1, end, trainDataList, K);
			invokeAll(task1, task2);
			return;
		}
	}
}
